#include "zrpc/base/timer_queue.h"

#include "zrpc/base/logger.h"
#include "zrpc/net/event_loop.h"

namespace zrpc {
#ifdef __APPLE__
TimerQueue::TimerQueue(EventLoop *loop_) : loop_(loop_) {}
#endif

#ifdef __linux__
int64_t CreateTimerfd() {
  int64_t timer_fd =
      ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
  if (timer_fd < 0) {
    LOG_FATAL << "CreateTimerfd error " << timer_fd;
  }
  return timer_fd;
}
#endif

int64_t HowMuchTimeFrom(const TimeStamp &when) {
  int64_t microseconds =
      when.GetMicroSecondsSinceEpoch() - TimeStamp::NowMicros();
  if (microseconds < 1000) {
    microseconds = 1000;
  }
  return static_cast<int>(microseconds / 1000);
}

int64_t TimerQueue::GetTimeout() const {
  loop_->AssertInLoopThread();
  if (timers_.empty()) {
    return 1000;
  } else {
    return HowMuchTimeFrom(timers_.begin()->second->GetExpiration());
  }
}

#ifdef __linux__
struct timespec HowMuchTimeFromNow(const TimeStamp &when) {
  int64_t microseconds =
      when.GetMicroSecondsSinceEpoch() - TimeStamp::NowMicros();
  if (microseconds < 100) {
    microseconds = 100;
  }

  struct timespec ts;
  ts.tv_sec =
      static_cast<time_t>(microseconds / TimeStamp::kMicroSecondsPerSecond);
  ts.tv_nsec = static_cast<int64_t>(
      (microseconds % TimeStamp::kMicroSecondsPerSecond) * 1000);
  return ts;
}

void ResetTimerfd(int64_t timerfd_, const TimeStamp &expiration) {
  struct itimerspec new_value;
  struct itimerspec old_value;
  bzero(&new_value, sizeof new_value);
  bzero(&old_value, sizeof old_value);
  new_value.it_value = HowMuchTimeFromNow(expiration);
  ::timerfd_settime(timerfd_, 0, &new_value, &old_value);
}

void ReadTimerfd(int64_t timerfd_, const TimeStamp &now) {
  uint64_t howmany;
  ssize_t n = socket::Read(timerfd_, &howmany, sizeof howmany);
  assert(n == sizeof howmany);
}
#endif

#ifdef __linux__
TimerQueue::TimerQueue(EventLoop *loop_)
    : loop_(loop_),
      timerfd_(CreateTimerfd()),
      timerfd_channel_(loop_, timerfd_),
      calling_expired_timers_(false) {
  timerfd_channel_.SetReadCallback(std::bind(&TimerQueue::HandleRead, this));
  timerfd_channel_.EnableReading();
}
#endif

TimerQueue::~TimerQueue() {
#ifdef __linux__
  timerfd_channel_.DisableAll();
  timerfd_channel_.Remove();
  ::close(timerfd_);
#endif
}

std::shared_ptr<Timer> TimerQueue::AddTimer(double when, bool repeat,
                                            TimerCallback &&cb) {
  TimeStamp time(AddTime(TimeStamp::Now(), when));
  std::shared_ptr<Timer> timer(
      new Timer(std::move(cb), std::move(time), repeat, when));
  loop_->RunInLoop(std::bind(&TimerQueue::AddTimerInLoop, this, timer));
  return timer;
}

std::shared_ptr<Timer> TimerQueue::AddTimer(TimeStamp &&stamp, double when,
                                            bool repeat, TimerCallback &&cb) {
  std::shared_ptr<Timer> timer(
      new Timer(std::move(cb), std::move(stamp), repeat, when));
  loop_->RunInLoop(std::bind(&TimerQueue::AddTimerInLoop, this, timer));
  return timer;
}

void TimerQueue::CancelTimer(const std::shared_ptr<Timer> &timer) {
  loop_->RunInLoop(std::bind(&TimerQueue::CancelInloop, this, timer));
}

void TimerQueue::CancelInloop(const std::shared_ptr<Timer> &timer) {
  loop_->AssertInLoopThread();
  assert(timers_.size() == active_timers_.size());

  auto it = active_timers_.find(timer->GetSequence());
  if (it != active_timers_.end()) {
    auto iter = timers_.find(timer->GetWhen());
    while (iter != timers_.end()) {
      if (timer->GetSequence() == iter->second->GetSequence()) {
        timers_.erase(iter);
        break;
      } else {
        ++iter;
      }
    }
    active_timers_.erase(it);
  } else if (calling_expired_timers_) {
    canceling_timers_.insert(std::make_pair(timer->GetSequence(), timer));
  }
  assert(timers_.size() == active_timers_.size());
}

void TimerQueue::AddTimerInLoop(const std::shared_ptr<Timer> &timer) {
  loop_->AssertInLoopThread();
  bool earliest_changed = Insert(timer);
  if (earliest_changed) {
#ifdef __linux__
    ResetTimerfd(timerfd_, timer->GetExpiration());
#endif
  }
}

std::shared_ptr<Timer> TimerQueue::GetTimerBegin() {
  if (timers_.empty()) {
    return nullptr;
  }
  return timers_.begin()->second;
}

void TimerQueue::HandleRead() {
  loop_->AssertInLoopThread();
  assert(timers_.size() == active_timers_.size());
  TimeStamp now(TimeStamp::NowMicros());

#ifdef __linux__
  ReadTimerfd(timerfd_, now);
#endif
  GetExpired(now);

  calling_expired_timers_ = true;
  canceling_timers_.clear();

  for (auto &it : expired_) {
    it.second->Run();
  }

  calling_expired_timers_ = false;
  Reset(now);
}

bool TimerQueue::Insert(const std::shared_ptr<Timer> &timer) {
  loop_->AssertInLoopThread();
  assert(timers_.size() == active_timers_.size());

  bool earliest_changed = false;
  int64_t now_time = timer->GetExpiration().GetMicroSecondsSinceEpoch();
  auto it = timers_.begin();
  if (it == timers_.end() || now_time < it->first) {
    earliest_changed = true;
  }

  timers_.insert(std::make_pair(now_time, timer));
  active_timers_.insert(std::make_pair(timer->GetSequence(), timer));
  assert(timers_.size() == active_timers_.size());
  return earliest_changed;
}

void TimerQueue::Reset(const TimeStamp &now) {
  TimeStamp next_expire;
  for (auto &it : expired_) {
    if (it.second->GetRepeat() &&
        canceling_timers_.find(it.second->GetSequence()) ==
            canceling_timers_.end()) {
      it.second->Restart(now);
      Insert(it.second);
    }
  }

  expired_.clear();
  if (!timers_.empty()) {
    next_expire = timers_.begin()->second->GetExpiration();
  }

  if (next_expire.Valid()) {
#ifdef __linux__
    ResetTimerfd(timerfd_, next_expire);
#endif
  }
}

size_t TimerQueue::GetTimerSize() {
  loop_->AssertInLoopThread();
  assert(timers_.size() == active_timers_.size());
  return timers_.size();
}

void TimerQueue::GetExpired(const TimeStamp &now) {
  assert(timers_.size() == active_timers_.size());
  int64_t now_time = now.GetMicroSecondsSinceEpoch();
  auto end = timers_.lower_bound(now_time);
  assert(end == timers_.end() || now_time <= end->first);
  expired_.insert(timers_.begin(), end);
  timers_.erase(timers_.begin(), end);

  for (auto &it : expired_) {
    size_t n = active_timers_.erase(it.second->GetSequence());
    assert(n == 1);
    (void)n;
  }
  assert(timers_.size() == active_timers_.size());
}

}  // namespace zrpc